package com.atguigu.day09;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Example7 {
    public static void main(String[] args) {
        var env = StreamExecutionEnvironment.getExecutionEnvironment();
        var tableEnv = StreamTableEnvironment
                .create(env, EnvironmentSettings.newInstance().inStreamingMode().build());

        tableEnv
                .executeSql(
                        "CREATE TABLE clicks (" +
                                "`username` STRING," +
                                "`url` STRING," +
                                "`ts` TIMESTAMP(3)," +
                                // 表示ts是事件时间，最大延迟时间是3秒
                                "WATERMARK FOR `ts` AS `ts` - INTERVAL '3' SECONDS)" +
                                "  WITH (" +
                                "  'connector' = 'filesystem'," +
                                "  'path' = '/home/yuantuzhi/flinktutorial0926/src/main/resources/file1.csv'," +
                                "  'format' = 'csv'" +
                                "  )"
                );

        tableEnv
                .executeSql(
                        "CREATE TABLE result_table (" +
                                "`username` STRING," +
                                "`count` BIGINT," +
                                "`window_start_time` TIMESTAMP(3)," +
                                "`window_end_time` TIMESTAMP(3)" +
                                ") WITH (" +
                                "'connector' = 'print'" +
                                ")"
                );

        tableEnv
                .executeSql(
                        "INSERT INTO result_table" +
                                "  SELECT `username`, COUNT(`username`) AS `count`," +
                                "         TUMBLE_START(`ts`, INTERVAL '5' SECONDS) AS `window_start_time`," +
                                "         TUMBLE_END(`ts`, INTERVAL '5' SECONDS) AS `window_end_time`" +
                                "  FROM clicks GROUP BY `username`, TUMBLE(`ts`, INTERVAL '5' SECONDS)"
                );
    }
}
